package com.bw.medical.realtime.app.dwd;

import com.bw.medical.realtime.common.MedicalCommon;
import com.bw.medical.realtime.util.CreateEnvUtil;
import com.bw.medical.realtime.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/*
9.5 用户域用户注册事务事实表
 */

public class DwdUserRegister {
    public static void main(String[] args) {
        // TODO 1. 初始化表处理环境
        StreamExecutionEnvironment env = CreateEnvUtil.getStreamEnv(8086, "medical_dwd_user_user_register");
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        env.setParallelism(1);

        // TODO 2. 读取数据
        String topic = MedicalCommon.KAFKA_ODS_TOPIC;
        String groupId = "medical_dwd_user_user_register";
        CreateEnvUtil.createOdsTable(tableEnv, topic, groupId);

        // TODO 3. 筛选数据
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "    `data`['id'] `id`, \n" +
                "    `data`['create_time'] `register_time`, \n" +
                "    concat('*@',split_index(`data`['email'],'@',1)) `email`, \n" +
                "    concat(substr(`data`['telephone'],1,3),'********') `telephone`, \n" +
                "    `data`['username'] `username`\n" +
                "from topic_db\n" +
                "where `table`='user'\n" +
                "and `type`='insert'");
        tableEnv.createTemporaryView("result_table",resultTable);

        // TODO 4. 写入 Kafka 目标主题
        String sinkTopic = "medical_dwd_user_register";
        tableEnv.executeSql("create table " + sinkTopic + "(\n" +
                "`id` string,\n" +
                "`register_time` string,\n" +
                "`email` string,\n" +
                "`telephone` string,\n" +
                "`username` string\n" +
                ") " + KafkaUtil.getKafkaSinkDDL(sinkTopic));
        tableEnv.executeSql("insert into " + sinkTopic + " select * from result_table");
    }
}
